package com.sailors.common.task.provider.application.event.listener;

import com.alibaba.fastjson.JSON;
import com.sailors.common.task.provider.application.aggregation.IFileTaskAggregation;
import com.sailors.common.task.provider.domian.service.IFileTaskService;
import com.sailors.common.task.provider.faced.model.dto.execute.FileTaskFailedExecuteDto;
import com.sailors.common.task.provider.faced.model.dto.execute.RefreshProcessExecuteDto;
import com.sailors.common.task.provider.infra.design.common.util.TryCatchUtil;
import com.sailors.common.task.provider.infra.system.constant.Topics;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutionException;

/**
 * File task kafka lisntener
 *
 * @description:
 * @author: cyj
 * @date 2023 -02-22 00:19:27
 */
@Component
@Slf4j
public class FileTaskKafkaLisntener {

    @Autowired
    private IFileTaskAggregation fileTaskAggregation;

    @KafkaListener(topics = Topics.TASK_PROCESS)
    public void taskProcess(ConsumerRecord consumerRecord) {
        Object value = consumerRecord.value();
        log.info("监听任务进度变更数据：{}", value);
        RefreshProcessExecuteDto refreshProcessExecuteDto = JSON.parseObject(String.valueOf(value), RefreshProcessExecuteDto.class);
        TryCatchUtil.tryC(() -> fileTaskAggregation.refreshProcess(refreshProcessExecuteDto));
    }


    @KafkaListener(topics = Topics.FAILED_FILE_TASK)
    public void failedFileTask(ConsumerRecord consumerRecord) {
        Object value = consumerRecord.value();
        log.info("监听任务失败数据：{}", value);
        FileTaskFailedExecuteDto fileTaskFailedExecuteDto = JSON.parseObject(String.valueOf(value), FileTaskFailedExecuteDto.class);
        TryCatchUtil.tryC(() -> {
            try {
                fileTaskAggregation.failed(fileTaskFailedExecuteDto);
            } catch (ExecutionException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
    }
}
